package yuekao6.ods;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import yuekao6.util.KafkaUtil;

public class ReadKafka {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //）指定流处理环境以及检查点相关的设置，从kafka的tms_ods主题中读取，创建消费者对象，消费数据  封装为流，筛选订单和订单明细数据，对流中的数据类型进行转换
        // ，按照order_id进行分组，定义侧输出流标签 下单放到主流，支付成功、取消运单、揽收(接单)、发单 转运完成、派送成功、签收放到侧输出流，从主流中提取侧输出流
        // ，将不同流的数据写到kafka的不同主题中（5分）
        DataStreamSource<String> streamSource = env.addSource(KafkaUtil.kafkaSource("tms_ods"));
        //筛选订单和订单明细数据
        SingleOutputStreamOperator<String> orderinfodata = streamSource.filter(x -> JSON.parseObject(x).getString("table").equals("order_info"));
        SingleOutputStreamOperator<String> ordercargodata = streamSource.filter(x -> JSON.parseObject(x).getString("table").equals("order_cargo"));

//        orderinfodata.print();
//        ordercargodata.print();
        //对流中的数据类型进行转换
        SingleOutputStreamOperator<JSONObject> mapDS = orderinfodata.map(x -> JSON.parseObject(x));

        OutputTag<String> tag60020 = new OutputTag<String>("60020") {
        };
        OutputTag<String> tag60030 = new OutputTag<String>("60030") {
        };
        OutputTag<String> tag60040 = new OutputTag<String>("60040") {
        };
        OutputTag<String> tag60050 = new OutputTag<String>("60050") {
        };
        OutputTag<String> tag60060 = new OutputTag<String>("60060") {
        };
        OutputTag<String> tag60070 = new OutputTag<String>("60070") {
        };
        OutputTag<String> tag60080 = new OutputTag<String>("60080") {
        };
        OutputTag<String> tag60099 = new OutputTag<String>("60099") {
        };


        SingleOutputStreamOperator<String> process = mapDS.keyBy(x -> x.getJSONObject("data").getString("id")).process(new ProcessFunction<JSONObject, String>() {
            @Override
            public void processElement(JSONObject jsonObject, ProcessFunction<JSONObject, String>.Context context, Collector<String> collector) throws Exception {
                String status = jsonObject.getJSONObject("data").getString("status");
                if ("60010".equals(status)) {
                    collector.collect(jsonObject.toJSONString());
                }
                if ("60020".equals(status)) {
                    context.output(tag60020, jsonObject.toJSONString());
                }
                if ("60030".equals(status)) {
                    context.output(tag60030, jsonObject.toJSONString());
                }
                if ("60040".equals(status)) {
                    context.output(tag60040, jsonObject.toJSONString());
                }
                if ("60050".equals(status)) {
                    context.output(tag60050, jsonObject.toJSONString());
                }
                if ("60060".equals(status)) {
                    context.output(tag60060, jsonObject.toJSONString());
                }
                if ("60070".equals(status)) {
                    context.output(tag60070, jsonObject.toJSONString());
                }
                if ("60080".equals(status)) {
                    context.output(tag60080, jsonObject.toJSONString());
                }
                if ("60099".equals(status)) {
                    context.output(tag60099, jsonObject.toJSONString());
                }
            }
        });

//        process.print("60010:");
//        process.getSideOutput(tag60020).print("60020:");
//        process.getSideOutput(tag60030).print("60030:");
//        process.getSideOutput(tag60040).print("60040:");
//        process.getSideOutput(tag60050).print("60050:");
//        process.getSideOutput(tag60060).print("60060:");
//        process.getSideOutput(tag60070).print("60070:");
//        process.getSideOutput(tag60080).print("60080:");
//        process.getSideOutput(tag60099).print("60099:");

        process.addSink(KafkaUtil.kafkaSink("tag60010"));
        process.getSideOutput(tag60020).addSink(KafkaUtil.kafkaSink("tag60020"));
        process.getSideOutput(tag60030).addSink(KafkaUtil.kafkaSink("tag60030"));
        process.getSideOutput(tag60040).addSink(KafkaUtil.kafkaSink("tag60040"));
        process.getSideOutput(tag60050).addSink(KafkaUtil.kafkaSink("tag60050"));
        process.getSideOutput(tag60060).addSink(KafkaUtil.kafkaSink("tag60060"));
        process.getSideOutput(tag60070).addSink(KafkaUtil.kafkaSink("tag60070"));
        process.getSideOutput(tag60080).addSink(KafkaUtil.kafkaSink("tag60080"));
        process.getSideOutput(tag60099).addSink(KafkaUtil.kafkaSink("tag60099"));

        env.execute();
    }
}

